<?php

namespace randy\queue\connector;

use AMQPChannelException;
use AMQPConnectionException;
use AMQPEnvelope;
use AMQPExchangeException;
use AMQPQueue;
use AMQPQueueException;
use Exception;
use Ramsey\Uuid\Uuid;
use randy\queue\Connector;
use randy\queue\MyRedis;
use randy\queue\Worker;
use swoole_process;
use think\facade\Log;

class Amqp extends Connector {

	protected $redis;
	protected $connection;
	protected $channel;
	protected $exchange;

	protected $options = [
		'expire'   => 60,
		'host'     => '127.0.0.1',
		'port'     => 5672,
		'username' => 'guest',
		'password' => 'guest',
		'default'  => 'default'
		//        'timeout'    => 0
	];
	protected $child = 0;
	protected $maxProcesses = 5;

	/**
	 * Amqp constructor.
	 *
	 * @param array $options
	 *
	 * @throws AMQPConnectionException
	 * @throws AMQPExchangeException
	 */
	public function __construct(array $options = []) {

		pcntl_signal(SIGCHLD, [
			$this,
			"signal_handler"
		]);
		set_time_limit(0);
		ini_set('default_socket_timeout', -1);
		$this->redis = MyRedis::getInstance();
		if (!extension_loaded('sockets')) {
			throw new Exception('sockets扩展未安装');
		}
		if (!empty($options)) {
			$this->options = array_merge($this->options, $options);
		}

		$this->connection = new \AMQPConnection($this->options);
		if (!$this->connection->connect()) {
			echo "Cannot connect to the broker";
			exit();
		}
		//申明渠道
		$this->channel = new \AMQPChannel($this->connection);
		$this->exchange = new \AMQPExchange($this->channel);
	}

	/**
	 * 生产者, 将数据投递至队列
	 *
	 * @param        $job_name
	 * @param        $job
	 * @param string $data
	 * @param null   $queue
	 *
	 * @return bool
	 * @throws AMQPChannelException
	 * @throws AMQPConnectionException
	 * @throws AMQPExchangeException
	 * @throws AMQPQueueException
	 * @throws Exception
	 * @author : Randy_chen
	 * @Date   : 2020/9/2
	 * @Time   : 17:35
	 */
	public function push($job_name, $job, $data = '', $queue = null) : bool {
		$payload = $this->createPayload($job, $data, $job_name, $queue);
		$result = $this->prepareMsg(json_decode($payload, true));
		if ($result) {
			return $this->pushRaw($payload, $queue, 0);
		}
		return false;
	}

	/**
	 * 投递数据至消息队列
	 *
	 * @param string $payload
	 * @param int    $delay
	 * @param null   $queue
	 *
	 * @return bool
	 * @throws AMQPChannelException
	 * @throws AMQPConnectionException
	 * @throws AMQPExchangeException
	 * @throws AMQPQueueException
	 * @author : Randy_chen
	 * @Date   : 2020/9/2
	 * @Time   : 17:43
	 */
	public function pushRaw(string $payload = '', $queue = null, int $delay = 0) : bool {
		$queue_name = $this->getQueue($queue);
		//秒转毫秒
		$delay = $delay * 1000;
		$exchange_name = $queue_name . '_topic';
		$this->exchange->setType('x-delayed-message');
		$this->exchange->setFlags(AMQP_DURABLE); //持久
		$this->exchange->setName($exchange_name);
		$this->exchange->setArgument('x-delayed-type', 'direct');
		$this->exchange->declareExchange();
		$queue = new \AMQPQueue($this->channel);
		$queue->setName($queue_name ?: '');
		$queue->setFlags(AMQP_DURABLE);
		$queue->declareQueue();
		$routingKey = $queue_name;
		$queue->bind($exchange_name, $routingKey);
		return $this->exchange->publish($payload,
		                                $routingKey,
		                                AMQP_NOPARAM,
		                                [
			                                'delivery_mode' => 2,
			                                'headers'       => ['x-delay' => $delay]
		                                ]
		);
	}

	/**
	 * 延迟队列实现
	 *
	 * @param        $job_name
	 * @param        $job
	 * @param string $data
	 * @param int    $delay
	 *
	 * @param null   $queue
	 *
	 * @return bool
	 * @throws AMQPChannelException
	 * @throws AMQPConnectionException
	 * @throws AMQPExchangeException
	 * @throws AMQPQueueException
	 * @throws Exception
	 * @author : Randy_chen
	 * @Date   : 2020/9/2
	 * @Time   : 17:44
	 */
	public function later($job_name, $job, $data = '', $delay = 0, $queue = null) : bool {
		$res = $this->createPayload($job, $data, $job_name, $queue);
		$result = $this->prepareMsg(json_decode($res, true));
		if ($result) {
			return $this->pushRaw($data = '', $queue, $delay);
		}
		return false;
	}

	//pop消费模式
	public function pop($queue = null) {
		// TODO: Implement pop() method.
	}

	//amqp消费者模式

	/**
	 *
	 * @param null $queue
	 * @param int  $delay
	 * @param int  $maxTries
	 *
	 * @throws AMQPChannelException
	 * @throws AMQPConnectionException
	 * @throws AMQPExchangeException
	 * @throws AMQPQueueException
	 * @throws \AMQPEnvelopeException
	 * @author : Randy_chen
	 * @Date   : 2020/9/3
	 * @Time   : 10:38
	 */
	public function consume($queue = null, $delay = 66, $maxTries = 0) {
		$queue_name = $this->getQueue($queue);
		$exchange_name = 'queues_' . '_topic';
		$this->exchange->setName($exchange_name);
		$amqp_queue = new \AMQPQueue($this->channel);
		$amqp_queue->setName($queue_name);
		$routingKey = $queue_name;
		$amqp_queue->bind($exchange_name, $routingKey);
		$callback = function ($envelope, $amqp_queue) use ($queue, $maxTries, $delay) {
			$worker = new Worker();
			if ($envelope->getBody()) {
				$job = new \randy\queue\job\Amqp($this, $envelope->getBody(), $queue);
				//检测是否可以执行
				$status = $this->getJobStatus($job->msg_id());
				//当前任务已经执行或者已经执行成功,则不再重复执行
				if ($status === 0 || $status === 1) {
					$amqp_queue->ack($envelope->getDeliveryTag()); //手动进行应答
					return;
				}

				$res = $worker->process($job, $delay, $maxTries);
				$res && $amqp_queue->ack($envelope->getDeliveryTag()); //手动进行应答
			}
		};
		$amqp_queue->consume($callback);

	}


	/**
	 * 预发送消息
	 *
	 * @param array $prepareMsgData 预发送消息体
	 *
	 * @return bool
	 * @throws Exception
	 */
	public function prepareMsg(array $prepareMsgData) : bool {

		$msg_id = $prepareMsgData['msg_id'];
		$redis = $this->redis;
		//存数据，还要存时间，依据时间查找超时的任务
		try {
			$redis->hset("message_system", (string) $msg_id, json_encode($prepareMsgData));
			$redis->zAdd("message_system_time", strtotime($prepareMsgData['create_time']), (string) $msg_id);
			return true;
		} catch (Exception $e) {
			throw $e;
		}

	}

	/**
	 * 确认并且投递参数
	 *
	 * @param string $msg_id
	 * @param int    $flag
	 *
	 * @return false
	 * @throws AMQPChannelException
	 * @throws AMQPConnectionException
	 * @throws AMQPExchangeException
	 * @throws AMQPQueueException
	 * @throws \think\Exception
	 */
	public function confirmMsgToSend(string $msg_id, int $flag) {
		$data = $this->redis->hget("message_system", (string) $msg_id);
		if (!empty($data)) {
			$data = json_decode($data, true);
			$data['status'] = 2;

			if ($flag == 2) {
				//被消息恢复子系统投递的任务 ,非主动投递数据
				$data['attempts']++;
			}
			$data = json_encode($data);
			//发布消息到交换机当中,并且绑定好路由关系
			if ($this->redis->hset("message_system", (string) $msg_id, $data) == 0) {
				//投递消息至消息恢复系统
				return $this->pushRaw($data, $data['queue'], 0);
			}
		} else {
			throw new \think\Exception('消息数据不存在!');
		}

		return false;
	}

	/**
	 * 消息消费成功
	 *
	 * @param string $msg_id
	 *
	 * @return bool
	 * @throws Exception
	 */
	public function ackMsg(string $msg_id) : bool {
		//删除已确认消费的消息
		try {
			$this->redis->hdel("message_system", (string) $msg_id);
			$this->redis->zrem("message_system_time", (string) $msg_id);
			return true;
		} catch (Exception $e) {
			throw $e;
		}

	}


	/**
	 *设置任务状态
	 *
	 * @param $msg_id
	 * @param $status 0 执行者 1-执行成功 ,-1 执行失败
	 *
	 * @author : Randy_chen
	 * @Date   : 2020/9/3
	 * @Time   : 16:26
	 */
	public function setJobStatus($msg_id, $status) {
		$this->redis->hset("message_job", $msg_id, $status);
	}

	//查询任务状态
	public function getJobStatus($msg_id) {
		return $this->redis->hGet("message_job", $msg_id);
	}


	/**
	 * 重新发布任务
	 *
	 * @param string $queue
	 * @param string $payload
	 * @param int    $delay
	 * @param int    $attempts
	 *
	 * @return bool
	 * @throws Exception
	 */
	public function release($queue, $payload, $delay, $attempts) {
		$payload = $this->setMeta($payload, 'attempts', $attempts);
		$payload = $this->setMeta($payload, 'create_time', date('Y-m-d H:i:s'));
		$result = $this->prepareMsg(json_decode($payload, true));
		if ($result) {
			return $this->pushRaw($payload, $queue, $delay);
		}
		return false;
	}

	/**
	 * 删除任务
	 *
	 * @param string $queue
	 * @param string $job
	 *
	 * @return void
	 */
	public function deleteReserved($queue, $job) {
		$msg_id = json_decode($job, true)['msg_id'];
		$this->redis->hdel("message_system", (string) $msg_id);
		$this->redis->zrem("message_system_time", (string) $msg_id);
	}

	/**
	 *
	 * @param        $job
	 * @param string $data
	 * @param null   $job_name
	 * @param null   $queue
	 *
	 * @return string
	 * @throws Exception
	 * @author : Randy_chen
	 * @Date   : 2020/9/2
	 * @Time   : 23:07
	 */
	protected function createPayload($job, $data = '', $job_name = null, $queue = null) {
		$payload = $this->setMeta(parent::createPayload($job, $data), 'msg_id', $this->getRandomId());
		$payload = $this->setMeta($payload, 'queue', $queue);
		$payload = $this->setMeta($payload, 'job_name', $job_name);
		$payload = $this->setMeta($payload, 'create_time', date('Y-m-d H:i:s'));
		return $this->setMeta($payload, 'attempts', 1);
	}

	/**
	 * 随机id
	 *
	 * @return string
	 * @throws Exception
	 */
	protected function getRandomId() {
		return Uuid::uuid4()
		           ->toString();
	}

	/**
	 * 获取队列名
	 *
	 * @param string|null $queue
	 *
	 * @return string
	 */
	protected function getQueue($queue) {
		return 'queues_' . ($queue ?: $this->options['default']);
	}


}